package com.youxin.dataStream

import org.apache.flink.streaming.api.datastream.DataStreamUtils
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

import scala.collection.JavaConverters._

/**
  * 需要后面再详细看的东西
  * 1：连接器相关的
  * 2：transform
  * 3:dataSink
  * 4:迭代器
  * 5：容错以及控制延迟    checkpoint
  * addSource - Attach a new source function. For example, to read from Apache Kafka you can use addSource(new FlinkKafkaConsumer08<>(...)). See connectors for more details.
  */
class WindowWordCount {}
object WindowWordCount{
  def main(args: Array[String]): Unit = {
    // 创建本地的测试环境
    // val env = StreamExecutionEnvironment.createLocalEnvironment()
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream("localhost",9999)
    val count = text.flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(10))
      .sum(1)
    count.print()
    // 输出迭代器
    val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(count.javaStream).asScala
    // 开始吧
    env.execute("window stream workCount")

  }

}
